package org.apache.flink.runtime.cleanup;

import java.util.concurrent.Executor;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.metrics.groups.ConsumerManagerMetricGroup;
import org.apache.flink.runtime.resourceconsumer.runner.ResourceConsumerRunner;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.concurrent.RetryStrategy;

/**
 * @类名: DefaultResourceCleanerFactory
 * @说明: 默认的资源清理器工厂, 构建资源清理器
 *
 * @author   leehom
 * @Date	 2024年2月19日 下午2:05:28
 * 修改记录：
 *
 * @see 	 
 */
/**
 * {@code DispatcherResourceCleanerFactory} instantiates {@link ResourceCleaner} instances that
 * clean cleanable resources from the {@link org.apache.flink.runtime.dispatcher.Dispatcher}.
 *
 * <p>We need to handle the {@link JobManagerRunnerRegistry} differently due to a dependency between
 * closing the {@link org.apache.flink.runtime.jobmaster.JobManagerRunner} and the {@link
 * HighAvailabilityServices}. This is fixed in {@code FLINK-24038} using a feature flag to
 * enable/disable single leader election for all the {@code JobManager} components. We can remove
 * the priority cleanup logic after removing the per-component leader election.
 */
public class DefaultResourceCleanerFactory implements ResourceCleanerFactory {

    /** 属性标签*/
    private static final String JOB_MANAGER_RUNNER_REGISTRY_LABEL = "JobManagerRunnerRegistry";
    private static final String BLOB_SERVER_LABEL = "BlobServer";
    private static final String HA_SERVICES_LABEL = "HighAvailabilityServices";
    private static final String JOB_MANAGER_METRIC_GROUP_LABEL = "JobManagerMetricGroup";

    private final Executor cleanupExecutor;
    private final RetryStrategy retryStrategy;

    // 清理/关闭组件
    private final ResourceConsumerRunner resourceConsumerRunner;
    private final BlobServer blobServer;
    private final HighAvailabilityServices highAvailabilityServices;
    private final ConsumerManagerMetricGroup jobManagerMetricGroup;

    @VisibleForTesting
    public DefaultResourceCleanerFactory(
            Executor cleanupExecutor,
            RetryStrategy retryStrategy,
            ResourceConsumerRunner resourceConsumerRunner,
            BlobServer blobServer,
            HighAvailabilityServices highAvailabilityServices,
            ConsumerManagerMetricGroup jobManagerMetricGroup) {
        this.cleanupExecutor = Preconditions.checkNotNull(cleanupExecutor);
        this.retryStrategy = retryStrategy;
        this.resourceConsumerRunner = Preconditions.checkNotNull(resourceConsumerRunner);
        this.blobServer = Preconditions.checkNotNull(blobServer);
        this.highAvailabilityServices = Preconditions.checkNotNull(highAvailabilityServices);
        this.jobManagerMetricGroup = Preconditions.checkNotNull(jobManagerMetricGroup);
    }

    @Override
    public ResourceCleaner createLocalResourceCleaner(
            ComponentMainThreadExecutor mainThreadExecutor) {
        return DefaultResourceCleaner.forLocallyCleanableResources(
                        mainThreadExecutor, cleanupExecutor, retryStrategy)
                .withPrioritizedCleanup(JOB_MANAGER_RUNNER_REGISTRY_LABEL, resourceConsumerRunner)
                .withRegularCleanup(BLOB_SERVER_LABEL, blobServer)
                .withRegularCleanup(JOB_MANAGER_METRIC_GROUP_LABEL, jobManagerMetricGroup)
                .build();
    }

    @Override
    public ResourceCleaner createGlobalResourceCleaner(
            ComponentMainThreadExecutor mainThreadExecutor) {
        return DefaultResourceCleaner.forGloballyCleanableResources(
                        mainThreadExecutor, cleanupExecutor, retryStrategy)
                .withPrioritizedCleanup(
                        JOB_MANAGER_RUNNER_REGISTRY_LABEL,
                        ofLocalResource(resourceConsumerRunner))
                .withRegularCleanup(BLOB_SERVER_LABEL, blobServer)
                .withRegularCleanup(HA_SERVICES_LABEL, highAvailabilityServices)
                .withRegularCleanup(
                        JOB_MANAGER_METRIC_GROUP_LABEL, ofLocalResource(jobManagerMetricGroup))
                .build();
    }

    /**
     * A simple wrapper for the resources that don't have any artifacts that can outlive the {@link
     * org.apache.flink.runtime.dispatcher.Dispatcher}, but we still want to clean up their local
     * state when we terminate globally.
     *
     * @param localResource Local resource that we want to clean during a global cleanup.
     * @return Globally cleanable resource.
     */
    private static GloballyCleanableResource ofLocalResource(
            LocallyCleanableResource localResource) {
        return localResource::localCleanupAsync;
    }
}
